package auto.utils;

import auto.base.TestNgSpringContextTests;
import com.kipa.utils.PrintUtils;
import com.kipa.utils.PropertiesUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.stereotype.Service;

import java.util.Properties;

/**
 * @author 马伟龙（weilong.ma@yummy.tech）
 * @since 2021年05月12日
 */

@Service
public class KafkaService extends TestNgSpringContextTests {

    private static KafkaProducer<String, String> producer;
    private static final String DEFAULT = "DEFAULT";

    public void kafkaProduct(String topic, Object data) {
        String env = (String) globalCacheContext.getAttribute("common.env");
        String configFile = "application.properties";
        if (DEFAULT.equals(env)) {
            configFile = "application.properties";
        } else {
            configFile = "application-" + env + ".properties";
        }
        Properties properties = PropertiesUtils.loadProperties(configFile);
        String bootstrapServers = PropertiesUtils.getProperty(properties, "spring.kafka.bootstrap-servers");
        String acks = PropertiesUtils.getProperty(properties, "spring.kafka.producer.acks");
        int retries = Integer.parseInt(PropertiesUtils.getProperty(properties, "spring.kafka.producer.retries"));
        int batchSize = Integer.parseInt(PropertiesUtils.getProperty(properties, "spring.kafka.producer.batch-size"));
        int bufferMemory = Integer.parseInt(PropertiesUtils.getProperty(properties, "spring.kafka.producer.buffer-memory"));
        String keySerializer = PropertiesUtils.getProperty(properties, "spring.kafka.producer.key-serializer");
        String valueSerializer = PropertiesUtils.getProperty(properties, "spring.kafka.producer.value-serializer");
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("acks", acks);
        props.put("retries", retries);
        props.put("batch.size", batchSize);
        props.put("buffer.memory", bufferMemory);
        props.put("key.serializer", keySerializer);
        props.put("value.serializer", valueSerializer);
        producer = new KafkaProducer(props);
        producer.send(new ProducerRecord(topic, data));
        PrintUtils.println("Kafka发送成功! " + topic + " : " + data.toString());
        producer.close();
    }
}